package study.stream

import java.io.{BufferedReader, InputStreamReader}
import java.net.Socket
import java.nio.charset.StandardCharsets

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable
import scala.util.Random

/**
 * spark stream 实现准实时的wordcount
 *
 * @author zh
 * @date 2021/5/28 10:42
 */
object SparkStreamingWordCount {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkStreaming")
    // 第二个参数是采集周期
    val streamingContext = new StreamingContext(sparkConf, Seconds(1))

    // 设置检查点
    streamingContext.checkpoint("./checkpoint2")

    val stream = streamingContext.receiverStream(new MyReceiver)

    val map = stream.map((_, 1))


    val window = map.reduceByKeyAndWindow(
      { (x, y) => x + y }, // 加上新进入窗口的批次中的元素
      { (x, y) => x - y }, //移除离开窗口的老批次中的元素
      Seconds(10), //窗口时长
      Seconds(1) //滑动步长
    )

    window.print()

//    val wordCount = map.reduceByKey(_ + _)
//
//
//    wordCount.print()

//    rddQueue.map((_,1)).reduceByKey(_+_).print()
    // 启动采集器
    streamingContext.start()

    // 等待采集器停止
    streamingContext.awaitTermination()

  }

}

// 自定义数据源
class MyReceiver() extends Receiver[String](StorageLevel.MEMORY_ONLY){

  var isStop : Boolean = false
  // 启动时调用，作用是用来读取数据并将数据发送给spark
  override def onStart(): Unit = {
    new Thread("Socket Receiver") {
      override def run() {
        while(!isStop){
          val arrays = Array("z","h","a","n","g")


          // 生成字符
          val message = arrays(new Random().nextInt(4))

          store(message)
          Thread.sleep(1000)
        }

      }
    }.start()
  }


  // 停止
  override def onStop(): Unit = {
    isStop = true

  }
}
